package article

import (
	"context"
	"gitee.com/DonHz/basic-go/webook/internal/repository"
	"gitee.com/DonHz/basic-go/webook/pkg/logger"
	"gitee.com/DonHz/basic-go/webook/pkg/saramax"
	"github.com/IBM/sarama"
	"time"
)

type InteractiveReadEventBatchConsumer struct {
	client sarama.Client
	l logger.Logger
	repo repository.InteractiveRepository
}


func NewInteractiveReadEventBatchConsumer(client sarama.Client, l logger.Logger,
	repo repository.InteractiveRepository) *InteractiveReadEventBatchConsumer {
	return &InteractiveReadEventBatchConsumer{
		client: client,
		l: l,
		repo: repo,
	}
}

func (k *InteractiveReadEventBatchConsumer) Start() error {
	cg, err :=sarama.NewConsumerGroupFromClient("interactive", k.client)
	if err!=nil{
		return err
	}
	go func() {
		err:=cg.Consume(context.Background(),
			[]string{"article_read"},
			saramax.NewBatchConsumerHandler[ReadEvent](k.l, k.Consume))
		if err!=nil{
			k.l.Error("退出消费循环异常", logger.Error(err))
		}
	}()
	return err
}

func (k *InteractiveReadEventBatchConsumer) Consume(msgs []*sarama.ConsumerMessage, ts []ReadEvent) error {
	ids:=make([]int64, 0, len(ts))
	bizs:=make([]string, 0, len(ts))
	for _, evt := range ts {
		ids = append(ids, evt.Aid)
		bizs = append(bizs, "article")
	}
	ctx, cancel:=context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	err:=k.repo.BatchIncrReadCnt(ctx, bizs, ids)
	if err!=nil{
		k.l.Error("批量增加阅读计数失败", logger.Error(err),
			logger.Field{Key: "ids", Value: ids})
	}
	return err
}